デフォルトVPCを全リージョンLambda(Python3)から削除する

鈴木純

2022.03.07

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

タイトルの通りですが、Lambdaから全リージョンのデフォルトVPCを削除してみたので、サンプルとしてコードを置いておきます。ご自由にお使いください。

そのほかにもデフォルトVPCを削除するブログは出てますので、用途によって使い分けて頂ければ幸いです。

サンプルコード

先にLambdaで利用するコードを載せておきます。LambdaのランタイムはPython3.8で動作確認をしています。

ログの出力や削除後のリソースチェックはしていますが、あくまでサンプルとしてご利用ください。

デフォルトVPCには追加のリソースが作成されていないことを前提としています。セキュリティグループ等がデフォルトVPCに紐づいている場合、失敗しますのでご了承ください。

デフォルトVPCを全リージョンで削除する

import boto3
import logging
from typing import List

logger = logging.getLogger()
logger.setLevel(logging.INFO)

def describe_regions() -> list:
    try:
        ec2 = boto3.client('ec2')
        regions = list(map(lambda x: x['RegionName'], ec2.describe_regions()['Regions']))
        return regions
    except Exception as e:
        logger.error(e)
        raise

def delete_sub(ec2, vpc_id: str):
    response = ec2.describe_subnets( \
        Filters=[ { 'Name': 'vpc-id', 'Values':[ vpc_id ], } ] )
    subs = map(lambda x:x['SubnetId'], response['Subnets'])
    for sub in subs:
        ec2.delete_subnet(SubnetId=sub)
    logger.info(f" Succeeded to delete Subnet")

def delete_igw(ec2, vpc_id: str):
    response = ec2.describe_internet_gateways( \
        Filters=[ { 'Name': 'attachment.vpc-id', 'Values':[ vpc_id ], } ] )

    for igw in response['InternetGateways']:
        for attach in igw['Attachments']:
            if attach['State'] == 'available':
                ec2.detach_internet_gateway(
                    InternetGatewayId = igw['InternetGatewayId'],
                    VpcId = vpc_id)
        ec2.delete_internet_gateway(InternetGatewayId=igw['InternetGatewayId'])
    logger.info(f" Succeeded to delete InternetGateway")

def delete_vpc(ec2, vpc_id: str):
    ec2.delete_vpc(VpcId=vpc_id)
    logger.info(f" Succeeded to delete VPC")

def get_default_vpc_id(ec2) -> str:
    attributes = ec2.describe_account_attributes(AttributeNames = ['default-vpc'] )
    vpc_id=attributes['AccountAttributes'][0]['AttributeValues'][0]['AttributeValue']
    return vpc_id

def is_get_default_vpc (regions: List[str]) -> bool:
    check_regions = {}
    for region_name in regions:
        ec2 = boto3.client('ec2',region_name=region_name)
        try:
            vpc_id=get_default_vpc_id(ec2)
            if vpc_id == "none":
                check_regions[region_name] = True
            else:
                check_regions[region_name] = False
        except Exception as e:
            logger.error(f"{e}:{region_name}")
            check_regions[region_name] = False
    return all(check_regions.values())
    
def lambda_handler(event, context):
    
    #全リージョン取得
    regions=describe_regions()

    errs = []
    for region_name in regions:
        try:
            ec2 = boto3.client('ec2',region_name=region_name)
            vpc_id=get_default_vpc_id(ec2)
            
            if vpc_id != "none":
                delete_sub(ec2, vpc_id)
                delete_igw(ec2, vpc_id)
                delete_vpc(ec2, vpc_id)
                logger.info(f"Succeeded to delete default VPC@{region_name}:{vpc_id}")
            else:
                logger.info(f"default VPC is already been deleted@{region_name}")
        except Exception as e:
            logger.error(f'Failed to delete default VPC@{region_name}')
            logger.error(e)
            errs.append(e)

    if errs:
        raise Exception('Failed to delete default VPC')
    
    is_resources = {
        "default_vpc": False,
    }
    is_resources["default_vpc"] = is_get_default_vpc(regions)

    # is_resourcesにFalseがないか確認
    if not all(is_resources.values()):
        logger.error("Failed to delete default vpc")
        logger.error(
            f"is_resources: {is_resources}"
        )
        raise Exception('Fail to check reource')

    logger.info("Completed to delete default vpc")
    logger.info(f"is_resources: {is_resources}")
    return {"is_resources": is_resources}

デフォルトVPCを全リージョンで作成する

実際に利用する際にはテストのために再作成したいこともあると思うので、全リージョンデフォルトVPCを作成できるコードも置いておきます。こちらはリソースの状態チェックまではしてません。

import boto3
import logging

logger = logging.getLogger()
logger.setLevel(logging.INFO)

def describe_regions() -> list:
    try:
        ec2 = boto3.client('ec2')
        regions = list(map(lambda x: x['RegionName'], ec2.describe_regions()['Regions']))
        return regions
    except Exception as e:
        logger.error(e)
        raise
def create_default_vpc(regions):
    for region in regions:
        ec2 = boto3.client('ec2', region_name=region)
        try:
            ec2.create_default_vpc()
            logger.info(f'Succeeded in create default vpc@{region}')
        except Exception as e:
            if e.response['Error']['Code'] == 'DefaultVpcAlreadyExists':
                logger.info(f'default vpc is alredy created@{region}')
            else:
                logger.info(f'Failed in create default vpc@{region}')
                logger.error(e)
        
def lambda_handler(event, context):

    #全リージョン取得
    regions=describe_regions()
    #デフォルトVPC作成
    create_default_vpc(regions)